package net.wicp.tams.common.binlog.self.bean;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.collections.CollectionUtils;

import com.alibaba.fastjson.annotation.JSONField;

import lombok.AccessLevel;
import lombok.Data;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.apiext.JdbcUtil;
import net.wicp.tams.common.apiext.StringUtil;
import net.wicp.tams.common.binlog.self.constant.ParamsVar;
import net.wicp.tams.common.binlog.self.constant.SenderEnum;
import net.wicp.tams.common.binlog.self.sender.ISender;
import net.wicp.tams.common.constant.dic.YesOrNo;

/***
 * 表
 * 
 * @author zhoujunhui
 *
 */
@Data
@Slf4j
public class Task {
	private String id;// 唯一标识
	private String ip;// ip地址或机器名
	private int clientId;// 连到mysql的从服务器ID
	private int port;
	private String dbinst;// 数据库实例名
	private String user;
	private String pwd;
	private String defaultDb;
	private String dbPattern;// db的模式
	private String tbPattern;// table的模式
	private String senderOther;
	private String senderEnum;
	private Map<String, String> params;
	// private String[] cols;
	private String remark;
	private YesOrNo multiThread;// 是否启用多线程
	private int threadNum;// 线程数，对于kafka将不起作用
	private int querySize;// 每线程队列大小
	private int tableNum;// 此实例有多少张表
	private String splitKey;

	@Setter(value = AccessLevel.PRIVATE)
	private List<ISender> senders;

	@JSONField(serialize = false)
	public List<ISender> getSenders() {
		this.senders = new ArrayList<>();
		if (StringUtil.isNull(this.senderEnum)
				|| ("no".equals(this.senderEnum) && StringUtil.isNull(this.senderOther))) {
			return this.senders;
		}
		String senderClass = "";
		if ("no".equals(senderEnum)) {
			senderClass = this.senderOther;
		} else {
			senderClass = this.senderEnum;
		}

		if (CollectionUtils.isEmpty(senders) && StringUtil.isNotNull(senderClass)) {
			String[] senderAry = senderClass.split(",");
			senders = new ArrayList<>();
			Map<String, String> paramsInput = new HashMap<>();
			if (params != null) {
				paramsInput.putAll(params);
			}
			paramsInput.put(ParamsVar.MULTITHREAD, multiThread == null ? "no" : multiThread.name());
			paramsInput.put(ParamsVar.THREADNUM, String.valueOf(threadNum));
			paramsInput.put(ParamsVar.QUERYSIZE, String.valueOf(querySize));
			paramsInput.put(ParamsVar.SPLITKEY, splitKey);
			paramsInput.put(ParamsVar.TABLENUM, String.valueOf(tableNum));
			for (String senderStr : senderAry) {
				try {
					SenderEnum temp = SenderEnum.get(senderStr);
					ISender sender = (ISender) Class.forName(temp == null ? senderStr : temp.getValue()).newInstance();
					sender.init(log, id, paramsInput);
					senders.add(sender);
				} catch (Exception e) {
					log.error(String.format("添加sender：[%s]错误", senderStr), e);
				}
			}
		}
		return senders;
	}

	@JSONField(serialize = false)
	public Pos getMastStatus() {
		String url;
		if (StringUtil.isNull(defaultDb)) {
			url = String.format("jdbc:mysql://%s:%s?autoReconnect=true&useUnicode=true&characterEncoding=utf-8", ip,
					port);
		} else {
			url = String.format("jdbc:mysql://%s:%s/%s?autoReconnect=true&useUnicode=true&characterEncoding=utf-8", ip,
					port, defaultDb);
		}
		java.sql.Connection conn = JdbcUtil.getConnection("com.mysql.jdbc.Driver", url, user, pwd);
		ResultSet rs = JdbcUtil.querySql(conn, "show master status");
		try {
			if (rs.next()) {
				String filename = rs.getString(1);
				long pos = rs.getLong(2);
				Pos ret = new Pos();
				ret.setFileName(filename);
				ret.setPos(pos);
				if (rs.getMetaData().getColumnCount() >= 5) {
					String gtidStr = rs.getString(5);
					ret.setGtids(gtidStr.replace("/n", ""));
				}
				rs = JdbcUtil.querySql(conn, "show variables like 'server_id'");
				if (rs.next()) {
					long masterServerId = rs.getLong(2);
					ret.setMasterServerId(masterServerId);
				}
				return ret;
			}
		} catch (SQLException e) {
			e.printStackTrace();
		} finally {
			try {
				rs.close();
				conn.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
		return null;
	}

	/**
	 * 为后面的热加载sender做准备
	 * 
	 * @param sender
	 */
	public void addSender(ISender sender) {
		if (senders == null) {
			senders = new ArrayList<>();
		}
		senders.add(sender);
	}
}
